package com.chengv.redismq.core;

import cn.hutool.core.lang.Assert;
import com.chengv.redismq.constant.RedisMQConstant;
import com.chengv.redismq.pojo.MQMessage;
import com.chengv.redismq.util.MethodUtils;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.stream.StreamListener;

import java.lang.reflect.Method;

/**
 * 消息监听器
 *
 * @author chengv
 * @see RedisMQStreamListenerRegistrar
 */
@Slf4j
public class RedisMQStreamListener implements StreamListener<String, ObjectRecord<String, String>> {

    @Getter
    private final Object listenerBean;
    @Getter
    private final Method listenerMethod;
    private final RedisTemplate<String, String> redisTemplate;
    private final RedisSerializer<Object> hashValueSerializer;
    @Getter
    private final String topic;
    @Getter
    private final String streamKey;
    @Getter
    private final String consumerGroup;

    public RedisMQStreamListener(Object listenerBean, Method listenerMethod, RedisTemplate<String, String> redisTemplate,
                                 RedisSerializer<Object> hashValueSerializer, String topic, String consumerGroup) {
        Assert.isTrue(listenerMethod.getParameterCount() == 1,
                String.format("监听器方法(%s)有且只能有一个参数", MethodUtils.getFullName(listenerBean, listenerMethod)));
        Assert.notBlank(topic, "topic 不能为空");
        Assert.notBlank(consumerGroup, "consumerGroup 不能为空");

        this.redisTemplate = redisTemplate;
        this.hashValueSerializer = hashValueSerializer;
        this.listenerBean = listenerBean;
        this.listenerMethod = listenerMethod;
        this.topic = topic;
        this.streamKey = RedisMQConstant.QUEUE_PREFIX + topic;
        this.consumerGroup = consumerGroup;
    }

    @Override
    public void onMessage(ObjectRecord<String, String> message) {
        try {
            log.info("[RedisMQStreamListener#onMessage][接收消息], listener=({}), message=({})", MethodUtils.getFullName
                    (listenerBean, listenerMethod), message);
            MQMessage mqMessage = deserializeToMQMessage(message.getValue());
            Object payloadObj = mqMessage.getPayload();
            // 调用消费者的方法
            listenerMethod.invoke(listenerBean, payloadObj);
            // ack 确认消费
            redisTemplate.opsForStream().acknowledge(consumerGroup, message);
        } catch (Exception e) {
            log.info("[RedisMQStreamListener#onMessage][消费失败], listener=({}), message=({})", MethodUtils.getFullName
                    (listenerBean, listenerMethod), message);
            throw new RuntimeException(e);
        }
    }

    /**
     * 反序列化 JSON 消息到 {@link MQMessage} 对象
     *
     * @param messageValue JSON 消息
     * @return 结果
     */
    private MQMessage deserializeToMQMessage(String messageValue) {
        return (MQMessage) hashValueSerializer.deserialize(messageValue.getBytes());
    }
}
